package com.cike.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class HdfsClient {

    /**
     * 新建文件夹
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */

    @Test
    public void testMkdirs() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        // 配置在集群上运行
        configuration.set("fs.defaultFS", "hdfs://atguigu-100:9000");
        //FileSystem fs = FileSystem.get(configuration);
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 创建目录
        fs.mkdirs(new Path("/1227/daxian/banzhang"));
        // 3 关闭资源
        fs.close();
    }

    /**
     * 删除文件夹
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testDelete() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 执行删除
        fs.delete(new Path("/1228/hadoop-2.7.2.tar.gz"), true);
        // 3 关闭资源
        fs.close();
    }

    /**
     * 上传文件测试
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        configuration.set("dfs.replication", "1");
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 上传文件
        fs.copyFromLocalFile(new Path("D:/devcodes/sublimeText3/atguigu.txt"), new Path("/1227/atguigu.txt"));
        // 3 关闭资源
        fs.close();
        System.out.println("over");
    }

    /**
     * 文件下载测试
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new
                URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 执行下载操作
        // boolean delSrc 指是否将原文件删除
        // Path src 指要下载的文件路径
        // Path dst 指将文件下载到的路径
        // boolean useRawLocalFileSystem 是否开启文件校验
        fs.copyToLocalFile(false, new Path("/1227/atguigu.txt"), new Path("C:/Users/cikelao/Desktop/atguigu.txt"), true);
        // 3 关闭资源
        fs.close();
    }

    /**
     * 测试重命名
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testRename() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 修改文件名称
        fs.rename(new Path("/1227/atguigu.txt"), new Path("/1227/atguigu2.txt"));
        // 3 关闭资源
        fs.close();
    }

    /**
     * 查看文件名称、权限、长度、块信息
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testListFiles() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 获取文件详情
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/1227/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus status = listFiles.next();
            // 输出详情
            // 文件名称
            System.out.println(status.getPath().getName());
            // 长度
            System.out.println(status.getLen());
            // 权限
            System.out.println(status.getPermission());
            // 分组
            System.out.println(status.getGroup());
            // 获取存储的块信息
            BlockLocation[] blockLocations = status.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                // 获取块存储的主机节点
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println("主机地址\t" + host);
                }
            }
            System.out.println("-----------输出分割线----------");
        }
        // 3 关闭资源
        fs.close();
    }

    /**
     * HDFS 文件和文件夹判断
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testListStatus() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件配置信息
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new
                URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 判断是文件还是文件夹
        FileStatus[] listStatus = fs.listStatus(new Path("/1227/"));
        for (FileStatus fileStatus : listStatus) {
            // 如果是文件
            if (fileStatus.isFile()) {
                System.out.println("f:"+fileStatus.getPath().getName());
            }else {
                System.out.println("d:"+fileStatus.getPath().getName());
            }
        }
        // 3 关闭资源
        fs.close();
    }

    /**
     *
     *  分割线--------------------------------------------
     *
     * */

    /**
     * 自己封装数据流上传输数据到HDFS
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 创建输入流
        FileInputStream fis = new FileInputStream(new File("D:/installpack/hadoop-2.7.2.tar.gz"));
        // 3 获取输出流
        FSDataOutputStream fos = fs.create(new Path("/1228/hadoop-2.7.2.tar.gz"));
        // 4 流对拷
        IOUtils.copyBytes(fis, fos, configuration);
        // 5 关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }

    /**
     * 从 HDFS 上下载 atguigu2.txt 文件到本地
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    // 文件下载
    @Test
    public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new
                URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 获取输入流
        FSDataInputStream fis = fs.open(new Path("/1228/hadoop-2.7.2.tar.gz"));
        // 3 获取输出流
        FileOutputStream fos = new FileOutputStream(new File("C:/Users/cikelao/Desktop/hadoop-2.7.2.tar.gz"));
        // 4 流的对拷
        IOUtils.copyBytes(fis, fos, configuration);
        // 5 关闭资源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
    }

    /**
     * 分块读取 HDFS 上的大文件，比如根目录下的/hadoop-2.7.2.tar.gz
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 获取输入流
        FSDataInputStream fis = fs.open(new Path("/1228/hadoop-2.7.2.tar.gz"));
        // 3 创建输出流
        FileOutputStream fos = new FileOutputStream(new File("C:/Users/cikelao/Desktop/hadoop-2.7.2.tar.gz.part1"));
        // 4 流的拷贝
        byte[] buf = new byte[1024];
        for(int i =0 ; i < 1024 * 128; i++){
            fis.read(buf);
            fos.write(buf);
        }
        // 5 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();
    }

    /**
     * 分块读取 HDFS 上的大文件，比如根目录下的/hadoop-2.7.2.tar.gz
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException{
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://atguigu-100:9000"), configuration, "atguigu");
        // 2 打开输入流
        FSDataInputStream fis = fs.open(new Path("/1228/hadoop-2.7.2.tar.gz"));
        // 3 定位输入数据位置
        fis.seek(1024*1024*128);
        // 4 创建输出流
        FileOutputStream fos = new FileOutputStream(new File("C:/Users/cikelao/Desktop/hadoop-2.7.2.tar.gz.part2"));
        // 5 流的对拷
        IOUtils.copyBytes(fis, fos, configuration);
        // 6 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    }

}
